package com.atguigu.gmall.realtime.app

import java.{lang, util}
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{DauInfo, OrderDetail, OrderInfo, OrderWide}
import com.atguigu.gmall.realtime.util.{MyEsUtil, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object DwdOrderWideApp {


  // 0   环境初始
  //1  加载kafka流 X2
  //2  提取偏移量 X2
  //3  转换结构 X2
  //4 维度关联  分别关联
  //5 双流join
  //6  写入es
  def main(args: Array[String]): Unit = {

    // 0   环境初始
    val sparkConf: SparkConf = new SparkConf().setAppName("dwd_order_wide_app").setMaster("local[4]")

    val ssc = new StreamingContext(sparkConf,Seconds(5))

    val orderInfoTopic="DWD_ORDER_INFO_I"
    val orderDetailTopic="DWD_ORDER_DETAIL_I"
    val groupId="dwd_order_wide_app"


    //1  加载kafka流 X2
    //主表
    val orderInfoOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderInfoTopic,groupId)
    val orderInputDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(orderInfoTopic,ssc,orderInfoOffsetMap,groupId)
   //从表
   val orderDetailOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderDetailTopic,groupId)
    val orderDetailDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(orderDetailTopic,ssc,orderDetailOffsetMap,groupId)

    //2  提取偏移量 X2
    //主表
    var orderInfoOffsetRanges: Array[OffsetRange]=null
    val orderInfoInputWithOffsetDstream: DStream[ConsumerRecord[String, String]] = orderInputDstream.transform { rdd =>
      //4个分区 的偏移量各种值
      orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //dr
      rdd
    }
    //从表
    var orderDetailOffsetRanges: Array[OffsetRange]=null
    val orderDetailInputWithOffsetDstream: DStream[ConsumerRecord[String, String]] = orderDetailDstream.transform { rdd =>
      //4个分区 的偏移量各种值
      orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //dr
      rdd
    }

    //3  转换结构 X2
    //主表
    val orderInfoDStream: DStream[OrderInfo] = orderInfoInputWithOffsetDstream.map { record =>
      val jsonString: String = record.value()
        val orderInfo: OrderInfo = JSON.parseObject(jsonString,classOf[OrderInfo])
       orderInfo
    }
    //从表
    val orderDetailDStream: DStream[OrderDetail] = orderDetailInputWithOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val orderDetail: OrderDetail = JSON.parseObject(jsonString,classOf[OrderDetail])
      orderDetail
    }


    //4 维度关联  分别关联      orderDetail  维度 作为作业
    val orderInfoWithDimDstream: DStream[OrderInfo] = orderInfoDStream.mapPartitions { orderInfoItr =>

      val jedis: Jedis = RedisUtil.getJedisClient
      val orderInfoList = new ListBuffer[OrderInfo]
      for (orderInfo <- orderInfoItr) {

        //补充用户信息
        // 性别
        val userInfoKey = s"DIM:USER_INFO:${orderInfo.user_id}"
        val userInfoJson: String = jedis.get(userInfoKey)
        val userJsonObj: JSONObject = JSON.parseObject(userInfoJson)
        val gender: String = userJsonObj.getString("gender")
        var genderCHN = ""
        if (gender == "M") {
          genderCHN = "男"
        } else if (gender == "F") {
          genderCHN = "女"
        } else {
          genderCHN = "未知"
        }
        orderInfo.user_gender = genderCHN
        // 年龄
        val birthday: String = userJsonObj.getString("birthday")
        val dateFormat = new SimpleDateFormat("yyyy-MM-dd")
        val birthdayDt: Date = dateFormat.parse(birthday)
        val diffTs = new Date().getTime - birthdayDt.getTime
        val age: Long = diffTs / 1000 / 3600 / 24 / 365
        orderInfo.user_age = age.toInt
        // 地区
        val provinceKey = s"DIM:BASE_PROVINCE:${orderInfo.province_id}"
        val provinceJson: String = jedis.get(provinceKey)
        val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)

        orderInfo.province_name = provinceJsonObj.getString("name")
        orderInfo.province_3166_2_code = provinceJsonObj.getString("iso_3166_2")
        orderInfo.province_iso_code = provinceJsonObj.getString("iso_code")
        orderInfo.province_area_code = provinceJsonObj.getString("area_code")

        //时间
        val dateTimeArr: Array[String] = orderInfo.create_time.split(" ")
        orderInfo.create_date=dateTimeArr(0)
        orderInfo.create_hour=dateTimeArr(1).split(":")(0)

        orderInfoList.append(orderInfo)



      }
      jedis.close()
      orderInfoList.toIterator
    }
   // orderInfoWithDimDstream.print(1000)

  //  orderDetailDStream.print(1000)

    //5 双流join
    val orderInfoWithKeyDstream: DStream[(Long, OrderInfo)] = orderInfoWithDimDstream.map(orderInfo=>(orderInfo.id,orderInfo))
    val orderDetailWithKeyDstream: DStream[(Long, OrderDetail)] = orderDetailDStream.map(orderDetail=>(orderDetail.order_id,orderDetail))
   // val orderJoinedDstream: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoWithKeyDstream.join(orderDetailWithKeyDstream)

    val orderJoinedDStream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoWithKeyDstream.fullOuterJoin(orderDetailWithKeyDstream)
   // val orderWideDStream: DStream[OrderWide]
    val orderWideDstream: DStream[OrderWide] = orderJoinedDStream.flatMap { case (orderId, (orderInfoOpt, orderDetailOpt)) =>
      val jedis: Jedis = RedisUtil.getJedisClient
      //1  主表 在
      val orderWideList = new ListBuffer[OrderWide]()
      if (orderInfoOpt != None) {
        val orderInfo: OrderInfo = orderInfoOpt.get
        // 1.1  判断从表在不在   从表在  生成新的对象   new OrderWide(orderInfo,orderDetail)
        if (orderDetailOpt != None) {
          val orderDetail: OrderDetail = orderDetailOpt.get
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        }
        //  1.2  主表写缓存
        //type ?  string   1:1  1好拆分 2 好过期     key?   ORDER_WIDE:ORDER_INFO:[id]    field/score?   无 value ?   orderInfoJson
        //写？ setex   读？  get    过期？ 1 小时

        val orderInfoKey = s"ORDER_WIDE:ORDER_INFO:${orderInfo.id}"
        val orderInfoJson: String = JSON.toJSONString(orderInfo, new SerializeConfig(true))
        jedis.setex(orderInfoKey, 3600, orderInfoJson)

        //   1.3 主表查从表缓存 如果从表在  生成新的对象
        //type ?    set     key?  ORDER_WIDE:ORDER_DETAIL:[order_id]    field/score?    无 value ?  orderDetailJson 多个
        //写？ sadd     读？ smemebers         过期？ 1 小时
        val orderDetailKey = s"ORDER_WIDE:ORDER_DETAIL:${orderInfo.id}"
        val orderDetailJsonSet: util.Set[String] = jedis.smembers(orderDetailKey)
        if (orderDetailJsonSet != null && orderDetailJsonSet.size() > 0) {
          import scala.collection.JavaConverters._
          for (orderDetailJson <- orderDetailJsonSet.asScala) {
            val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
            val orderWide = new OrderWide(orderInfo, orderDetail)
            orderWideList.append(orderWide)
          }
        }
      } else {
        // 2   主表不在 从表 在
        val orderDetail: OrderDetail = orderDetailOpt.get

        //2.1 从表查主表缓存   如果主表在  生成新的对象
        val orderInfoKey = s"ORDER_WIDE:ORDER_INFO:${orderDetail.order_id}"
        val orderInfoJson: String = jedis.get(orderInfoKey)
        if (orderInfoJson != null) {
          val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        } else { // 2.2  如果没查到主表 从表写缓存
          val orderDetailkey = s"ORDER_WIDE:ORDER_DETAIL:${orderDetail.order_id}"
          val orderDetailJson: String = JSON.toJSONString(orderDetail, new SerializeConfig(true))
          jedis.sadd(orderDetailkey, orderDetailJson)
          jedis.expire(orderDetailkey, 3600)
        }


      }
      jedis.close()
      orderWideList

    }

    orderWideDstream.cache()
    orderWideDstream.print(1000)



    //orderJoinedDstream.print(1000)



    //6  写入es
    orderWideDstream.foreachRDD{rdd=>
      rdd.foreachPartition{ orderWideItr=>

        val orderWideList: List[(String, OrderWide)] = orderWideItr.toList.map(orderWide=>(orderWide.detail_id.toString,orderWide))
        if(orderWideList!=null&& orderWideList.size>0){
          val orderWideFirst: (String, OrderWide) = orderWideList(0)
          val orderWideDt: String = orderWideFirst._2.create_date
          val  indexName:String =s"gmall0819_order_wide_$orderWideDt"
          MyEsUtil.bulkSave(orderWideList,indexName)
        }
      }
      //dr
      OffsetManager.saveOffset(orderInfoTopic,groupId,orderInfoOffsetRanges)
      OffsetManager.saveOffset(orderDetailTopic,groupId,orderDetailOffsetRanges)

    }

    ssc.start()
    ssc.awaitTermination()


  }

}
